package org.platform.lwc.system.listener.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.platform.lwc.kafka.model.KafkaMessage;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/**
 * description: 消费
 *
 * @author lwc
 * @date 2022/4/10
 */
@Component
@Slf4j
public class ConsumerListener {

    /**
     * @Description: 建议看一下KafkaListener的源码 很多api 我们也可以指定分区消费消息
     *  topicPartitions ={@TopicPartition(topic = "topic1", partitions = { "0", "1" })}
     */
    @KafkaListener(topics = "demo-group", groupId = "eshop-cache-group")
    public void listen(String message, Acknowledgment ack) throws JsonProcessingException {
        log.info("消息内容" + message);
        //手动提交offset
        ack.acknowledge();
        log.info("消费结束");
    }

}

